package com.ceair.util;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;

import java.util.concurrent.CompletableFuture;

/**
 * @author wangbaohai
 * @ClassName KafkaUtils
 * @description: Kafka工具类
 * @date 2025年04月17日
 * @version: 1.0.0
 */
@Slf4j
@Component
public class KafkaUtils {

    private final KafkaTemplate<String, String> kafkaTemplate;

    public KafkaUtils(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    /**
     * 发送消息到指定的Kafka主题。
     *
     * @param topic 主题名称，指定消息将发送到的Kafka主题。
     * @param messageJson 消息内容，以JSON字符串的形式表示，将作为Kafka消息的值发送。
     *
     * 该方法通过异步方式发送消息，并在发送完成后执行回调处理结果或异常。
     * 如果发送失败，会记录错误日志，并预留了持久化补偿机制的扩展点。
     */
    public void sendKafkaMessage(String topic, String messageJson) {
        try {
            // 创建ProducerRecord对象，指定Kafka主题和消息内容
            ProducerRecord<String, String> record = new ProducerRecord<>(topic, messageJson);

            // 使用KafkaTemplate异步发送消息，并获取发送结果的Future对象
            CompletableFuture<SendResult<String, String>> future = kafkaTemplate.send(record);

            // 注册回调函数，处理消息发送完成后的结果或异常
            future.whenComplete((result, ex) -> {
                if (ex == null) {
                    // 如果发送成功，记录消息内容及其在Kafka中的偏移量
                    log.info("Sent message=[{}] with offset=[{}]", record.value(), result.getRecordMetadata().offset());
                } else {
                    // 如果发送失败，记录错误日志并提示需要添加持久化补偿机制
                    log.error("Unable to send message=[{}] due to : {}", record.value(), ex.getMessage());
                    // TODO 这里需要添加持久化补偿，但是目前方案待定，之后再加
                }
            });
        } catch (Exception e) {
            // 捕获异常并记录日志，避免异常继续抛出，等待后续持久化补偿机制处理
            log.error("Send Kafka Message Failed, topic:{}, message:{}, reason:{}", topic, messageJson, e.toString());
        }
    }

}
